package knowlwdge_graph.knowledge_etl.yago_etl;

import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * Created by julianzliu on 4/15/2017.
 */
public class TransformerP2O {
    /*
    transformer of postgres to orientdb
     */

    static final org.slf4j.Logger logger = LoggerFactory.getLogger(TransformerP2O.class);

    private ExtractorPostgres extractorPostgres;
    private LoderOrientDB loderOrientDB;

    public TransformerP2O(){
        this.extractorPostgres = new ExtractorPostgres();
        this.loderOrientDB = new LoderOrientDB();
    }


    public RowYagoTriple transform(RowYagoTriple row){
        //this.loderOrientDB.saveTripleTuple( row );
        return row;
    }


    public void traversalAll(int begin, int end, int batch_size){
        int offset = begin;
        int limit = batch_size;
        int ret_size = 0;
        while (true){
            if(offset >= end){
                logger.info("[traversalAll] success, begin  = {}, end = {}, offset = {}", begin, end, offset);
                break;
            }
            if(offset + batch_size > end )
                limit = end - offset;
            else limit = batch_size;

            List<RowYagoTriple> rows = this.extractorPostgres.getPartitialRowSet(offset, limit);
            ret_size = rows.size();
            logger.info("[traversalAll] offset = {}, ret_size = {}", offset, ret_size);
            if(ret_size == 0) break;

            for(RowYagoTriple row : rows){
                RowYagoTriple turple = transform( row );
                /*
                多线程可能会冲突，需要多次重试机制
                 */
                int retry = 3;
                while(retry > 0){
                    try{
                        this.loderOrientDB.saveTripleTuple( turple );
                        break;
                    }catch (Exception e){
                        if(retry > 0) retry -= 1;
                        else{
                            logger.error("[Retry Failed!] msg = \n{}", e.getMessage());
                            try{
                                Thread.sleep(1000);
                            }catch (Exception e2){
                                logger.error("[Sleep Failed!] msg = \n{}", e2.getMessage());
                            }
                        }
                    }
                }

            }

            offset += ret_size;
            this.extractorPostgres.storeLatestOffset(offset, 1);
            Integer latest_offset = this.extractorPostgres.getLatestOffset();
            System.out.println("latest offset = " + latest_offset);
            try{
                Thread.sleep(1000);
            }catch (Exception e){
                e.printStackTrace();
            }

        }
    }

    public static class TransformerThread implements Runnable {

        private TransformerP2O transformer = new TransformerP2O();
        private int begin;
        private int end;
        private int batch_size;

        public TransformerThread(int begin, int end, int batch_size){
            super();
            this.begin = begin;
            this.end = end;
            this.batch_size = batch_size;
            logger.info("\n\n[TransformerThread] begin = {}, end = {}, batch_size = {}\n\n", begin, end, batch_size);
        }

        public void run(){
            this.transformer.traversalAll(this.begin, this.end, this.batch_size);
        }
    }


    /*****************************************************************
     * MAIN
     */
    public static void main(String[] args){
        int batch_size = Integer.parseInt(args[0]);
        TransformerP2O transformer = new TransformerP2O();
        transformer.traversalAll(0, 100000000, batch_size);
    }
}
